package sheep.rxjava.rxtest;


import cn.hutool.core.thread.ThreadUtil;
import io.reactivex.Observable;
import io.reactivex.Single;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class JavaTest {

    public static Logger log = LoggerFactory.getLogger(JavaTest.class);


    @Test
    // 简单使用
    public void test() {
        // testSingle();
        testRetry();
    }

    private void testRetry() {
        Single.create((emitter) -> {
            log.info("执行 emitter  ");
            int i = 1 / 0;
            emitter.onSuccess(11);
        }).subscribeOn(Schedulers.single())
                .retry(3)
                .subscribe(
                        i -> log.info("onNext: {}", i),
                        ex -> log.error("onError: {}", ex.getMessage())
                );

        ThreadUtil.sleep(3000);
    }

    private void testSingle() {
        log.info("执行主线程 : {} ", Thread.currentThread().getName());
        Disposable subscribe = Observable.create(emitter -> {
            log.info(" 执行 subscribe  , 线程 : {} ", Thread.currentThread().getName());
            ThreadUtil.sleep(2000);
            // int i = 1 / 0;
            emitter.onNext(1);
        })
                .subscribeOn(Schedulers.io())  // 指定上游事件 , 异步线程中 , 执行事件
                .observeOn(Schedulers.io())  //来决定下游事件被处理的线程
                .subscribe(
                        resu -> {
                            log.info("next : {} , 线程 : {}  ", resu, Thread.currentThread().getName());
                        },
                        err -> {
                            log.error("onError : {}  , 线程 : {}   ", err.getMessage(), Thread.currentThread().getName());
                        }

                );

        log.info("isDisposed : {}", subscribe.isDisposed());
        ThreadUtil.sleep(3000);
    }


}
